/*-------------------------------------------------------------------------
 *
 * vacuumlo.c
 *	  移除数据库中孤立的大对象。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  contrib/vacuumlo/vacuumlo.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres_fe.h"

#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#ifdef HAVE_TERMIOS_H
#include <termios.h>
#endif

#include "catalog/pg_class_d.h"
#include "common/connect.h"
#include "common/logging.h"
#include "common/string.h"
#include "getopt_long.h"
#include "libpq-fe.h"
#include "pg_getopt.h"

#define BUFSIZE			1024

enum trivalue
{
	TRI_DEFAULT,
	TRI_NO,
	TRI_YES
};

struct _param
{
	char	   *pg_user;
	enum trivalue pg_prompt;
	char	   *pg_port;
	char	   *pg_host;
	const char *progname;
	int			verbose;
	int			dry_run;
	long		transaction_limit;
};

static int	fc_vacuumlo(const char *fc_database, const struct _param *fc_param);
static void fc_usage(const char *fc_progname);



/*
 * 这将清理一个数据库的 LO。成功时返回 0，失败时返回 -1。
 */
static int fc_vacuumlo(const char *fc_database, const struct _param *fc_param)
{
	PGconn	   *fc_conn;
	PGresult   *fc_res,
			   *fc_res2;
	char		fc_buf[BUFSIZE];
	long		fc_matched;
	long		fc_deleted;
	int			fc_i;
	bool		fc_new_pass;
	bool		fc_success = true;
	static char *fc_password = NULL;

	/* 注意：密码可以从之前的调用中携带过来 */
	if (fc_param->pg_prompt == TRI_YES && !fc_password)
		fc_password = simple_prompt("Password: ", false);

	/*
	 * 开始连接。循环直到我们获得后端请求的密码。
	 */
	do
	{
#define PARAMS_ARRAY_SIZE	   7

		const char *fc_keywords[PARAMS_ARRAY_SIZE];
		const char *fc_values[PARAMS_ARRAY_SIZE];

		fc_keywords[0] = "host";
		fc_values[0] = fc_param->pg_host;
		fc_keywords[1] = "port";
		fc_values[1] = fc_param->pg_port;
		fc_keywords[2] = "user";
		fc_values[2] = fc_param->pg_user;
		fc_keywords[3] = "password";
		fc_values[3] = fc_password;
		fc_keywords[4] = "dbname";
		fc_values[4] = fc_database;
		fc_keywords[5] = "fallback_application_name";
		fc_values[5] = fc_param->progname;
		fc_keywords[6] = NULL;
		fc_values[6] = NULL;

		fc_new_pass = false;
		fc_conn = PQconnectdbParams(fc_keywords, fc_values, true);
		if (!fc_conn)
		{
			pg_log_error("connection to database \"%s\" failed", fc_database);
			return -1;
		}

		if (PQstatus(fc_conn) == CONNECTION_BAD &&
			PQconnectionNeedsPassword(fc_conn) &&
			!fc_password &&
			fc_param->pg_prompt != TRI_NO)
		{
			PQfinish(fc_conn);
			fc_password = simple_prompt("Password: ", false);
			fc_new_pass = true;
		}
	} while (fc_new_pass);

	/* 检查以确保后端连接成功建立 */
	if (PQstatus(fc_conn) == CONNECTION_BAD)
	{
		pg_log_error("%s", PQerrorMessage(fc_conn));
		PQfinish(fc_conn);
		return -1;
	}

	if (fc_param->verbose)
	{
		fprintf(stdout, "Connected to database \"%s\"\n", fc_database);
		if (fc_param->dry_run)
			fprintf(stdout, "Test run: no large objects will be removed!\n");
	}

	fc_res = PQexec(fc_conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
	if (PQresultStatus(fc_res) != PGRES_TUPLES_OK)
	{
		pg_log_error("failed to set search_path: %s", PQerrorMessage(fc_conn));
		PQclear(fc_res);
		PQfinish(fc_conn);
		return -1;
	}
	PQclear(fc_res);

	/*
	 * 首先我们创建并填充 LO 临时表。
	 */
	fc_buf[0] = '\0';
	strcat(fc_buf, "CREATE TEMP TABLE vacuum_l AS ");
	if (PQserverVersion(fc_conn) >= 90000)
		strcat(fc_buf, "SELECT oid AS lo FROM pg_largeobject_metadata");
	else
		strcat(fc_buf, "SELECT DISTINCT loid AS lo FROM pg_largeobject");
	fc_res = PQexec(fc_conn, fc_buf);
	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		pg_log_error("failed to create temp table: %s", PQerrorMessage(fc_conn));
		PQclear(fc_res);
		PQfinish(fc_conn);
		return -1;
	}
	PQclear(fc_res);

	/*
	 * 分析临时表，以便规划器为下面的 DELETE 生成合理的计划。
	 */
	fc_buf[0] = '\0';
	strcat(fc_buf, "ANALYZE vacuum_l");
	fc_res = PQexec(fc_conn, fc_buf);
	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		pg_log_error("failed to vacuum temp table: %s", PQerrorMessage(fc_conn));
		PQclear(fc_res);
		PQfinish(fc_conn);
		return -1;
	}
	PQclear(fc_res);

	/*
	 * 现在查找具有 oid 类型列的候选表。
	 *
	 * 注意：我们通过拒绝名为 'pg_*' 的模式中的表来忽略系统表和临时表。
	 * 特别是，上面形成的临时表被忽略，pg_largeobject 也会被忽略。如果
	 * 扫描了这些表，显然我们将没有任何内容可供删除...
	 */
	fc_buf[0] = '\0';
	strcat(fc_buf, "SELECT s.nspname, c.relname, a.attname ");
	strcat(fc_buf, "FROM pg_class c, pg_attribute a, pg_namespace s, pg_type t ");
	strcat(fc_buf, "WHERE a.attnum > 0 AND NOT a.attisdropped ");
	strcat(fc_buf, "      AND a.attrelid = c.oid ");
	strcat(fc_buf, "      AND a.atttypid = t.oid ");
	strcat(fc_buf, "      AND c.relnamespace = s.oid ");
	strcat(fc_buf, "      AND t.typname in ('oid', 'lo') ");
	strcat(fc_buf, "      AND c.relkind in (" CppAsString2(RELKIND_RELATION) ", " CppAsString2(RELKIND_MATVIEW) ")");
	strcat(fc_buf, "      AND s.nspname !~ '^pg_'");
	fc_res = PQexec(fc_conn, fc_buf);
	if (PQresultStatus(fc_res) != PGRES_TUPLES_OK)
	{
		pg_log_error("failed to find OID columns: %s", PQerrorMessage(fc_conn));
		PQclear(fc_res);
		PQfinish(fc_conn);
		return -1;
	}

	for (fc_i = 0; fc_i < PQntuples(fc_res); fc_i++)
	{
		char	   *fc_schema,
				   *fc_table,
				   *fc_field;

		fc_schema = PQgetvalue(fc_res, fc_i, 0);
		fc_table = PQgetvalue(fc_res, fc_i, 1);
		fc_field = PQgetvalue(fc_res, fc_i, 2);

		if (fc_param->verbose)
			fprintf(stdout, "Checking %s in %s.%s\n", fc_field, fc_schema, fc_table);

		fc_schema = PQescapeIdentifier(fc_conn, fc_schema, strlen(fc_schema));
		fc_table = PQescapeIdentifier(fc_conn, fc_table, strlen(fc_table));
		fc_field = PQescapeIdentifier(fc_conn, fc_field, strlen(fc_field));

		if (!fc_schema || !fc_table || !fc_field)
		{
			pg_log_error("%s", PQerrorMessage(fc_conn));
			PQclear(fc_res);
			PQfinish(fc_conn);
			if (fc_schema != NULL)
				PQfreemem(fc_schema);
			if (fc_table != NULL)
				PQfreemem(fc_table);
			if (fc_field != NULL)
				PQfreemem(fc_field);
			return -1;
		}

		snprintf(fc_buf, BUFSIZE,
				 "DELETE FROM vacuum_l "
				 "WHERE lo IN (SELECT %s FROM %s.%s)",
				 fc_field, fc_schema, fc_table);
		fc_res2 = PQexec(fc_conn, fc_buf);
		if (PQresultStatus(fc_res2) != PGRES_COMMAND_OK)
		{
			pg_log_error("failed to check %s in table %s.%s: %s",
						 fc_field, fc_schema, fc_table, PQerrorMessage(fc_conn));
			PQclear(fc_res2);
			PQclear(fc_res);
			PQfinish(fc_conn);
			PQfreemem(fc_schema);
			PQfreemem(fc_table);
			PQfreemem(fc_field);
			return -1;
		}
		PQclear(fc_res2);

		PQfreemem(fc_schema);
		PQfreemem(fc_table);
		PQfreemem(fc_field);
	}
	PQclear(fc_res);

	/*
	 * 现在，vacuum_l 中剩下的条目是孤立的。删除它们。
	 *
	 * 我们不想将每个删除作为单独的事务执行，因为提交开销会很高。
	 * 然而，自 9.0 开始，后端将为每个删除的 LO 获取一个锁，因此每个事务中
	 * 删除太多 LOs 会导致共享内存锁表中的空间不足。因此，我们每个事务
	 * 删除最多 transaction_limit 个 LOs。
	 */
	fc_res = PQexec(fc_conn, "begin");
	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		pg_log_error("failed to start transaction: %s", PQerrorMessage(fc_conn));
		PQclear(fc_res);
		PQfinish(fc_conn);
		return -1;
	}
	PQclear(fc_res);

	fc_buf[0] = '\0';
	strcat(fc_buf,
		   "DECLARE myportal CURSOR WITH HOLD FOR SELECT lo FROM vacuum_l");
	fc_res = PQexec(fc_conn, fc_buf);
	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		pg_log_error("DECLARE CURSOR failed: %s", PQerrorMessage(fc_conn));
		PQclear(fc_res);
		PQfinish(fc_conn);
		return -1;
	}
	PQclear(fc_res);

	snprintf(fc_buf, BUFSIZE, "FETCH FORWARD %ld IN myportal",
			 fc_param->transaction_limit > 0 ? fc_param->transaction_limit : 1000L);

	fc_deleted = 0;

	do
	{
		fc_res = PQexec(fc_conn, fc_buf);
		if (PQresultStatus(fc_res) != PGRES_TUPLES_OK)
		{
			pg_log_error("FETCH FORWARD failed: %s", PQerrorMessage(fc_conn));
			PQclear(fc_res);
			PQfinish(fc_conn);
			return -1;
		}

		fc_matched = PQntuples(fc_res);
		if (fc_matched <= 0)
		{
			/* 结果集末尾 */
			PQclear(fc_res);
			break;
		}

		for (fc_i = 0; fc_i < fc_matched; fc_i++)
		{
			Oid			fc_lo = atooid(PQgetvalue(fc_res, fc_i, 0));

			if (fc_param->verbose)
			{
				fprintf(stdout, "\rRemoving lo %6u   ", fc_lo);
				fflush(stdout);
			}

			if (fc_param->dry_run == 0)
			{
				if (lo_unlink(fc_conn, fc_lo) < 0)
				{
					pg_log_error("failed to remove lo %u: %s", fc_lo,
								 PQerrorMessage(fc_conn));
					if (PQtransactionStatus(fc_conn) == PQTRANS_INERROR)
					{
						fc_success = false;
						break;	/* 离开内层for循环 */  
					}
				}
				else
					fc_deleted++;
			}
			else
				fc_deleted++;

			if (fc_param->transaction_limit > 0 &&
				(fc_deleted % fc_param->transaction_limit) == 0)
			{
				fc_res2 = PQexec(fc_conn, "commit");
				if (PQresultStatus(fc_res2) != PGRES_COMMAND_OK)
				{
					pg_log_error("failed to commit transaction: %s",
								 PQerrorMessage(fc_conn));
					PQclear(fc_res2);
					PQclear(fc_res);
					PQfinish(fc_conn);
					return -1;
				}
				PQclear(fc_res2);
				fc_res2 = PQexec(fc_conn, "begin");
				if (PQresultStatus(fc_res2) != PGRES_COMMAND_OK)
				{
					pg_log_error("failed to start transaction: %s",
								 PQerrorMessage(fc_conn));
					PQclear(fc_res2);
					PQclear(fc_res);
					PQfinish(fc_conn);
					return -1;
				}
				PQclear(fc_res2);
			}
		}

		PQclear(fc_res);
	} while (fc_success);

	/*
	 * 就这些，朋友们！
	 */
	fc_res = PQexec(fc_conn, "commit");
	if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		pg_log_error("failed to commit transaction: %s",
					 PQerrorMessage(fc_conn));
		PQclear(fc_res);
		PQfinish(fc_conn);
		return -1;
	}
	PQclear(fc_res);

	PQfinish(fc_conn);

	if (fc_param->verbose)
	{
		if (fc_param->dry_run)
			fprintf(stdout, "\rWould remove %ld large objects from database \"%s\".\n",
					fc_deleted, fc_database);
		else if (fc_success)
			fprintf(stdout,
					"\rSuccessfully removed %ld large objects from database \"%s\".\n",
					fc_deleted, fc_database);
		else
			fprintf(stdout, "\rRemoval from database \"%s\" failed at object %ld of %ld.\n",
					fc_database, fc_deleted, fc_matched);
	}

	return ((fc_param->dry_run || fc_success) ? 0 : -1);
}

static void fc_usage(const char *fc_progname)
{
	printf("%s removes unreferenced large objects from databases.\n\n", fc_progname);
	printf("Usage:\n  %s [OPTION]... DBNAME...\n\n", fc_progname);
	printf("Options:\n");
	printf("  -l, --limit=LIMIT         commit after removing each LIMIT large objects\n");
	printf("  -n, --dry-run             don't remove large objects, just show what would be done\n");
	printf("  -v, --verbose             write a lot of progress messages\n");
	printf("  -V, --version             output version information, then exit\n");
	printf("  -?, --help                show this help, then exit\n");
	printf("\nConnection options:\n");
	printf("  -h, --host=HOSTNAME       database server host or socket directory\n");
	printf("  -p, --port=PORT           database server port\n");
	printf("  -U, --username=USERNAME   user name to connect as\n");
	printf("  -w, --no-password         never prompt for password\n");
	printf("  -W, --password            force password prompt\n");
	printf("\n");
	printf("Report bugs to <%s>.\n", PACKAGE_BUGREPORT);
	printf("%s home page: <%s>\n", PACKAGE_NAME, PACKAGE_URL);
}


int main(int fc_argc, char **fc_argv)
{
	static struct option fc_long_options[] = {
		{"host", required_argument, NULL, 'h'},
		{"limit", required_argument, NULL, 'l'},
		{"dry-run", no_argument, NULL, 'n'},
		{"port", required_argument, NULL, 'p'},
		{"username", required_argument, NULL, 'U'},
		{"verbose", no_argument, NULL, 'v'},
		{"version", no_argument, NULL, 'V'},
		{"no-password", no_argument, NULL, 'w'},
		{"password", no_argument, NULL, 'W'},
		{"help", no_argument, NULL, '?'},
		{NULL, 0, NULL, 0}
	};

	int			fc_rc = 0;
	struct _param fc_param;
	int			fc_c;
	int			fc_port;
	const char *fc_progname;
	int			fc_optindex;

	pg_logging_init(fc_argv[0]);
	fc_progname = get_progname(fc_argv[0]);

	/* 设置默认参数值 */
	fc_param.pg_user = NULL;
	fc_param.pg_prompt = TRI_DEFAULT;
	fc_param.pg_host = NULL;
	fc_param.pg_port = NULL;
	fc_param.progname = fc_progname;
	fc_param.verbose = 0;
	fc_param.dry_run = 0;
	fc_param.transaction_limit = 1000;

	/* 处理命令行参数 */
	if (fc_argc > 1)
	{
		if (strcmp(fc_argv[1], "--help") == 0 || strcmp(fc_argv[1], "-?") == 0)
		{
			fc_usage(fc_progname);
			exit(0);
		}
		if (strcmp(fc_argv[1], "--version") == 0 || strcmp(fc_argv[1], "-V") == 0)
		{
			puts("vacuumlo (PostgreSQL) " PG_VERSION);
			exit(0);
		}
	}

	while ((fc_c = getopt_long(fc_argc, fc_argv, "h:l:np:U:vwW", fc_long_options, &fc_optindex)) != -1)
	{
		switch (fc_c)
		{
			case 'h':
				fc_param.pg_host = pg_strdup(optarg);
				break;
			case 'l':
				fc_param.transaction_limit = strtol(optarg, NULL, 10);
				if (fc_param.transaction_limit < 0)
					pg_fatal("transaction limit must not be negative (0 disables)");
				break;
			case 'n':
				fc_param.dry_run = 1;
				fc_param.verbose = 1;
				break;
			case 'p':
				fc_port = strtol(optarg, NULL, 10);
				if ((fc_port < 1) || (fc_port > 65535))
					pg_fatal("invalid port number: %s", optarg);
				fc_param.pg_port = pg_strdup(optarg);
				break;
			case 'U':
				fc_param.pg_user = pg_strdup(optarg);
				break;
			case 'v':
				fc_param.verbose = 1;
				break;
			case 'w':
				fc_param.pg_prompt = TRI_NO;
				break;
			case 'W':
				fc_param.pg_prompt = TRI_YES;
				break;
			default:
				/* getopt_long already emitted a complaint */
				pg_log_error_hint("Try \"%s --help\" for more information.", fc_progname);
				exit(1);
		}
	}

	/* 没有给数据库？显示用法 */
	if (optind >= fc_argc)
	{
		pg_log_error("missing required argument: database name");
		pg_log_error_hint("Try \"%s --help\" for more information.", fc_progname);
		exit(1);
	}

	for (fc_c = optind; fc_c < fc_argc; fc_c++)
	{
		/* 在选定的数据库上工作 */
		fc_rc += (fc_vacuumlo(fc_argv[fc_c], &fc_param) != 0);
	}

	return fc_rc;
}
